package com.dbh.alg.algorithm.bigdata;


import java.util.*;
import java.util.concurrent.*;

/**
 * 寻找热门查询，300万个查询字符串中统计最热门的10个查询
 */
public class TopNFrequentItems {

    /**
     * 统计出现次数最多的前N个数
     * @param data 输入数据列表
     * @param n 要返回的前n个数
     * @return 包含前N个出现次数最多数据的列表
     * @param <T> 数据类型
     */
    public static <T> List<T> topNFrequent(List<T> data, int n) throws ExecutionException, InterruptedException {
        // 使用hashMap统计频率
        Map<T, Integer> frequentMap = countFrequentConcurrent(data);

        // 使用最小堆获取频率最高的前n个元素
        PriorityQueue<Map.Entry<T, Integer>> minHeap = new PriorityQueue<>(Comparator.comparingInt(Map.Entry::getValue));
        for (Map.Entry<T, Integer> entry : frequentMap.entrySet()) {
            if (minHeap.size() < n) {
                minHeap.offer(entry);
            } else {
                if (entry.getValue() > minHeap.peek().getValue()) {
                    minHeap.poll();
                    minHeap.offer(entry);
                }
            }
        }

        // 将结果从高到低排序
        List<T> result = new ArrayList<>();
        while (!minHeap.isEmpty()) {
            result.add(minHeap.poll().getKey());
        }
        Collections.reverse(result);

        return result;
    }

    private static final int THREAD_COUNT = Runtime.getRuntime().availableProcessors();

    /**
     * 多线程统计频率
     * @param data 输入数据列表
     * @return 频率数据
     * @param <T> 数据类型
     */
    public static <T> Map<T, Integer> countFrequentConcurrent(List<T> data) throws ExecutionException, InterruptedException {
        // 创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
        // 将数据分片
        int batchSize = data.size() / THREAD_COUNT;
        List<Future<?>> futureList = new ArrayList<>();
        Map<T, Integer> frequentMap = new ConcurrentHashMap<>();
        for (int i = 0; i < THREAD_COUNT; i++) {
            int fromIndex = i * batchSize;
            int toIndex = (i == THREAD_COUNT - 1) ? data.size() : (i + 1) * batchSize;
            List<T> subList = data.subList(fromIndex, toIndex);
            futureList.add(executorService.submit(() -> countFrequent(subList, frequentMap)));
        }

        for (Future<?> future :futureList) {
            future.get();
        }

        executorService.shutdown();

        return frequentMap;
    }

    private static  <T> void countFrequent(List<T> data,  Map<T, Integer> frequentMap) {
        for (T item : data) {
            frequentMap.put(item, frequentMap.getOrDefault(item, 0) + 1);
        }
    }

    /**
     * 大数据量测试方法
     */
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 模拟生成1亿条数据(有重复)
        List<String> data = generateTestDataConcurrent(100_000_000);

        // 统计前10个最频繁的数据
        long startTime = System.currentTimeMillis();
        List<String> top10 = topNFrequent(data, 10);
        long endTime = System.currentTimeMillis();

        System.out.println("Top 10 most frequent items:");
        for (String item : top10) {
            System.out.println(item);
        }
        System.out.printf("Time taken: %d ms%n", endTime - startTime);
    }

    /**
     * 生成测试数据
     * @param size 数据量大小
     * @return 测试数据列表
     */
    private static List<String> generateTestDataConcurrent(int size) throws InterruptedException {
        int threadCount = Runtime.getRuntime().availableProcessors();
        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
        List<Future<List<String>>> futures = new ArrayList<>();
        List<String> result = Collections.synchronizedList(new ArrayList<>(size));

        // 每个线程处理的数据量
        int batchSize = size / threadCount;

        for (int i = 0; i < threadCount; i++) {
            final int from = i * batchSize;
            final int to = (i == threadCount - 1) ? size : (i + 1) * batchSize;

            futures.add(executor.submit(() -> {
                Random random = new Random();
                String[] commonItems = {"apple", "banana", "orange", "pear", "grape",
                        "watermelon", "pineapple", "mango", "strawberry", "blueberry"};
                List<String> localList = new ArrayList<>(to - from);

                for (int j = from; j < to; j++) {
                    if (random.nextDouble() < 0.1) {
                        localList.add(commonItems[random.nextInt(commonItems.length)]);
                    } else {
                        localList.add("item_" + random.nextInt(10000));
                    }
                }
                return localList;
            }));
        }

        // 合并结果
        for (Future<List<String>> future : futures) {
            try {
                result.addAll(future.get());
            } catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        }

        executor.shutdown();
        return result;
    }
}
